package cn.doitedu.olap_agg;

import cn.doitedu.udfs.GetNullUDF;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author: 深似海
 * @Site: <a href="www.51doit.com">多易教育</a>
 * @QQ: 657270652
 * @Date: 2024/5/11
 * @Desc: 学大数据，上多易教育
 *   广告olap分析聚合表开发
 **/
public class Job05_AdOlapAggregate {

    public static void main(String[] args) {


        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointStorage("file:///d:/ckpt");

        env.setParallelism(1);


        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        tenv.getConfig().set("table.exec.source.idle-timeout","1 ms");



        // 1. 创建kafka数据源表（行为日志明细宽表）
        tenv.executeSql(
                "create table dwd_events_kafka(     " +
                        "     user_id bigint             " +
                        "    ,event_id           string             " +
                        "    ,action_time        bigint             " +
                        "    ,properties         map<string,string> " +
                        "    ,creative_id as properties['creative_id'] " +   // 广告创意id
                        "    ,ad_location_id as properties['ad_location_id'] " +  // 广告栏位id
                        "    ,ad_tracking_id as properties['ad_tracking_id'] " +  // 广告过程跟踪id
                        "    ,province   string        " +
                        "    ,city       string        " +
                        "    ,region     string        " +
                        "    ,rt as to_timestamp_ltz(action_time,3)   " +
                        "    ,watermark for rt as rt - interval '0' second   " +
                        ") WITH (                      " +
                        "    'connector' = 'kafka',    " +
                        "    'topic' = 'dwd-user-action-log',  " +
                        "    'properties.bootstrap.servers' = 'doitedu:9092',  " +
                        "    'properties.group.id' = 'doit44_g1',              " +
                        "    'scan.startup.mode' = 'latest-offset',            " +
                        "    'value.format' = 'json',                          " +
                        "    'value.fields-include' = 'EXCEPT_KEY'             " +
                        ")                                                     ");


        // 建表，映射doris中的物理表
        tenv.executeSql(
                " create table doris_sink(                                                                    "+
                        "      dt   date                                                                                 "+
                        "     ,province  varchar(30)                                                                       "+
                        "     ,city      varchar(30)                                                                           "+
                        "     ,region    varchar(30)                                                                         "+
                        "     ,ad_location_id  varchar(30)                                                                 "+
                        "     ,creative_id     varchar(30)                                                                    "+
                        "     ,show_time  bigint                                                                            "+
                        "     ,show_cnt  bigint                                                                            "+
                        "     ,show_bm   bigint                                                                             "+
                        "     ,click_cnt bigint                                                                            "+
                        "     ,click_bm  bigint                                                                             "+
                        "     ,transfer_cnt  bigint                                                                        "+
                        "     ,transfer_bm   bigint                                                                          "+
                        " ) WITH (                                                                                         "+
                        "    'connector' = 'doris',                                                                        "+
                        "    'fenodes' = 'doitedu:8030',                                                                   "+
                        "    'table.identifier' = 'dws.ad_olap_agg',                                               "+
                        "    'username' = 'root',                                                                          "+
                        "    'password' = 'root',                                                                          "+
                        "    'sink.label-prefix' = 'doris_label-47-009',                                                       "+
                        "    'sink.properties.columns' = 'dt,province,city,region,ad_location_id,creative_id, show_time,   "+
                        "    show_cnt,show_bm,click_cnt,click_bm,transfer_cnt,transfer_bm,                                 "+
                        "    show_bm=to_bitmap(show_bm),click_bm=to_bitmap(click_bm), transfer_bm=to_bitmap(transfer_bm)'  "+
                        " )                                                                                                "
        );



        // 按照  省、市、区、广告栏位id、广告创意id, user_id => 分组:   统计曝光时间、曝光次数、曝光人数、点击次数、点击人数、转化次数、转化人数
        tenv.createTemporaryFunction("get_null", GetNullUDF.class);
        tenv.executeSql(
                " insert into doris_sink                                                                                       "+
                        " with tmp as (                                                                                                "+
                        "     select                                                                                                   "+
                        "     *                                                                                                        "+
                        "     from dwd_events_kafka                                                                                    "+
                        "     where event_id in ('ad_show','ad_click','ad_transfer')                                                   "+
                        " )                                                                                                            "+
                        "                                                                                                              "+
                        " select                                                                                                       "+
                        "      TO_DATE(date_format(to_timestamp_ltz(min(action_time),3),'yyyy-MM-dd')) as dt                           "+
                        "     ,province                                                                                                "+
                        "     ,city                                                                                                    "+
                        "     ,region                                                                                                  "+
                        "     ,ad_location_id                                                                                          "+
                        "     ,creative_id                                                                                             "+
                        "     ,min(action_time) as show_time                                                                           "+
                        "     ,count(event_id) filter(where event_id = 'ad_show') as show_cnt                                          "+
                        "     ,user_id as show_bm                                                                                      "+
                        "     ,count(event_id) filter(where event_id = 'ad_click') as click_cnt                                        "+
                        "     ,if(count(event_id) filter(where event_id = 'ad_click')>0,user_id ,get_null() ) as click_bm              "+
                        "     ,count(event_id) filter(where event_id = 'ad_transfer') as transfer_cnt                                  "+
                        "     ,if(count(event_id) filter(where event_id = 'ad_transfer')>0,user_id ,get_null()  ) as transfer_bm       "+
                        " from tmp                                                                                                     "+
                        " group by                                                                                                     "+
                        "     province,                                                                                                "+
                        "     city,                                                                                                    "+
                        "     region,                                                                                                  "+
                        "     ad_location_id,                                                                                          "+
                        "     creative_id,                                                                                             "+
                        "     user_id                                                                                                  "
        );







    }
}
